OPC Studio User's Guide and Reference
Examples - OPC UA PubSub - MQTT JSON mapping using TCP

.NET

// This example shows how to subscribe to all dataset messages on an OPC-UA PubSub connection with MQTT JSON mapping using
// TCP.
//
// The following package needs to be referenced in your project (or otherwise made available) for the MQTT transport to 
// work.
// - OpcLabs.MqttNet
// Refer to the documentation for more information.
//
// Find all latest examples here: https://opclabs.doc-that.com/files/onlinedocs/OPCLabs-OpcStudio/Latest/examples.html .

using System;
using System.Collections.Generic;
using System.Threading;
using OpcLabs.EasyOpc.UA.PubSub;
using OpcLabs.EasyOpc.UA.PubSub.Engine;
using OpcLabs.EasyOpc.UA.PubSub.OperationModel;

namespace UADocExamples.PubSub._EasyUASubscriber
{
    partial class SubscribeDataSet
    {
        public static void MqttJsonTcp()
        {
            // Define the PubSub connection we will work with. Uses implicit conversion from a string.
            // Default port with MQTT is 1883.
            UAPubSubConnectionDescriptor pubSubConnectionDescriptor = "mqtt://opcua-pubsub.demo-this.com";
            // Specify the transport protocol mapping.
            // The statement below isn't actually necessary, due to automatic message mapping recognition feature; see
            // https://kb.opclabs.com/OPC_UA_PubSub_Automatic_Message_Mapping_Recognition for more details.
            pubSubConnectionDescriptor.TransportProfileUriString = UAPubSubTransportProfileUriStrings.MqttJson;

            // Define the arguments for subscribing to the dataset, specifying the MQTT topic name.
            var subscribeDataSetArguments = new UASubscribeDataSetArguments(pubSubConnectionDescriptor)
            {
                DataSetSubscriptionDescriptor = {CommunicationParameters = {BrokerDataSetReaderTransportParameters =
                {
                    QueueName = "opcuademo/json"
                }}}
            };

            // Instantiate the subscriber object and hook events.
            var subscriber = new EasyUASubscriber();
            subscriber.DataSetMessage += subscriber_DataSetMessage_MqttJsonTcp;

            Console.WriteLine("Subscribing...");
            subscriber.SubscribeDataSet(subscribeDataSetArguments);

            Console.WriteLine("Processing dataset message events for 20 seconds...");
            Thread.Sleep(20 * 1000);

            Console.WriteLine("Unsubscribing...");
            subscriber.UnsubscribeAllDataSets();

            Console.WriteLine("Waiting for 1 second...");
            // Unsubscribe operation is asynchronous, messages may still come for a short while.
            Thread.Sleep(1 * 1000);

            Console.WriteLine("Finished.");
        }

        static void subscriber_DataSetMessage_MqttJsonTcp(object sender, EasyUADataSetMessageEventArgs e)
        {
            // Display the dataset.
            if (e.Succeeded)
            {
                // An event with null DataSetData just indicates a successful connection.
                if (!(e.DataSetData is null))
                {
                    Console.WriteLine();
                    Console.WriteLine($"Dataset data: {e.DataSetData}");
                    foreach (KeyValuePair<string, UADataSetFieldData> pair in e.DataSetData.FieldDataDictionary)
                        Console.WriteLine(pair);
                }
            }
            else
            {
                Console.WriteLine();
                Console.WriteLine($"*** Failure: {e.ErrorMessage}");
            }
        }

        // Example output:
        //
        //Subscribing...
        //Processing dataset message events for 20 seconds...
        //
        //Dataset data: 2020-01-21T17:07:19.778,836,700,00; Good; Data; publisher=[String]31, class=eae79794-1af7-4f96-8401-4096cd1d8908, fields: 4
        //[BoolToggle, True {System.Boolean} @2020-01-21T16:07:19.778,836,700,00; Good]
        //[Int32, 482 {System.Int64} @2020-01-21T16:07:19.778,836,700,00; Good]
        //[Int32Fast, 2287 {System.Int64} @2020-01-21T16:07:19.778,836,700,00; Good]
        //[DateTime, 1/21/2020 5:07:19 PM {System.DateTime} @2020-01-21T16:07:19.778,836,700,00; Good]
        //
        //Dataset data: Good; Data; publisher=[String]32, fields: 4
        //[BoolToggle, True {System.Boolean}; Good]
        //[Int32, 482 {System.Int32}; Good]
        //[Int32Fast, 2287 {System.Int32}; Good]
        //[DateTime, 1/21/2020 5:07:19 PM {System.DateTime}; Good]
        //
        //Dataset data: Good; Data; publisher=[String]32, fields: 100
        //[Mass_0, 82 {System.Int64}; Good]
        //[Mass_1, 182 {System.Int64}; Good]
        //[Mass_2, 282 {System.Int64}; Good]
        //[Mass_3, 382 {System.Int64}; Good]
        //[Mass_4, 482 {System.Int64}; Good]
        //[Mass_5, 582 {System.Int64}; Good]
        //[Mass_6, 682 {System.Int64}; Good]
        //[Mass_7, 782 {System.Int64}; Good]
        //[Mass_8, 882 {System.Int64}; Good]
        //[Mass_9, 982 {System.Int64}; Good]
        //[Mass_10, 1082 {System.Int64}; Good]
        ////...
    }
}
# This example shows how to subscribe to all dataset messages on an OPC-UA PubSub connection with MQTT JSON mapping using
# TCP.
#
# The following package needs to be referenced in your project (or otherwise made available) for the MQTT transport to
# work.
# - OpcLabs.MqttNet
# Refer to the documentation for more information.
#
# Find all latest examples here: https://opclabs.doc-that.com/files/onlinedocs/OPCLabs-OpcStudio/Latest/examples.html .
# OPC client and subscriber examples in Python on GitHub: https://github.com/OPCLabs/Examples-QuickOPC-Python .
# The QuickOPC package is needed. Install it using "pip install opclabs_quickopc".
import opclabs_quickopc
import opclabs_mqttnet
import time

# Import .NET namespaces.
from OpcLabs.EasyOpc.UA.PubSub import *
from OpcLabs.EasyOpc.UA.PubSub.Engine import *
from OpcLabs.EasyOpc.UA.PubSub.OperationModel import *


def dataSetMessage(sender, e):
    # Display the dataset.
    if e.Succeeded:
        # An event with null DataSetData just indicates a successful connection.
        if e.DataSetData is not None:
            print('')
            print('Dataset data: ', e.DataSetData, sep='')
            for pair in e.DataSetData.FieldDataDictionary:
                print(pair)
    else:
        print('')
        print('*** Failure: ', e.ErrorMessageBrief, sep='')
        print('*** Failure: ', e.Exception, sep='')


# Define the PubSub connection we will work with. Uses implicit conversion from a string.
# Default port with MQTT is 1883.
pubSubConnectionDescriptor = UAPubSubConnectionDescriptor.op_Implicit('mqtt://opcua-pubsub.demo-this.com')
# Specify the transport protocol mapping.
# The statement below isn't actually necessary, due to automatic message mapping recognition feature; see
# https://kb.opclabs.com/OPC_UA_PubSub_Automatic_Message_Mapping_Recognition for more details.
pubSubConnectionDescriptor.TransportProfileUriString = UAPubSubTransportProfileUriStrings.MqttJson

# Define the arguments for subscribing to the dataset, specifying the MQTT topic name.
subscribeDataSetArguments = UASubscribeDataSetArguments(pubSubConnectionDescriptor)
subscribeDataSetArguments.DataSetSubscriptionDescriptor.CommunicationParameters.BrokerDataSetReaderTransportParameters.\
    QueueName = 'opcuademo/json'

# Instantiate the subscriber object and hook events.
subscriber = EasyUASubscriber()
subscriber.DataSetMessage += dataSetMessage

print('Subscribing...')
IEasyUASubscriberExtension.SubscribeDataSet(subscriber, subscribeDataSetArguments)

print('Processing dataset message events for 20 seconds...')
time.sleep(20)

print('Unsubscribing...')
subscriber.UnsubscribeAllDataSets()

print('Waiting for 1 second...')
# Unsubscribe operation is asynchronous, messages may still come for a short while.
time.sleep(1)

subscriber.DataSetMessage -= dataSetMessage

print('Finished.')
' This example shows how to subscribe to all dataset messages on an OPC-UA PubSub connection with MQTT JSON mapping using
' TCP.
'
' The following package needs to be referenced in your project (or otherwise made available) for the MQTT transport to 
' work.
' - OpcLabs.MqttNet
' Refer to the documentation for more information.
'
' Find all latest examples here: https://opclabs.doc-that.com/files/onlinedocs/OPCLabs-OpcStudio/Latest/examples.html .

Imports OpcLabs.EasyOpc.UA.PubSub
Imports OpcLabs.EasyOpc.UA.PubSub.Engine
Imports OpcLabs.EasyOpc.UA.PubSub.OperationModel

Namespace PubSub._EasyUASubscriber
    Partial Friend Class SubscribeDataSet
        Public Shared Sub MqttJsonTcp()

            ' Define the PubSub connection we will work with. Uses implicit conversion from a string.
            ' Default port with MQTT is 1883.
            Dim pubSubConnectionDescriptor As UAPubSubConnectionDescriptor = "mqtt://opcua-pubsub.demo-this.com"

            ' Specify the transport protocol mapping.
            ' The statement below isn't actually necessary, due to automatic message mapping recognition feature; see
            ' https://kb.opclabs.com/OPC_UA_PubSub_Automatic_Message_Mapping_Recognition for more details.
            pubSubConnectionDescriptor.TransportProfileUriString = UAPubSubTransportProfileUriStrings.MqttJson

            ' Define the arguments for subscribing to the dataset, specifying the MQTT topic name.
            Dim subscribeDataSetArguments = New UASubscribeDataSetArguments(pubSubConnectionDescriptor)
            subscribeDataSetArguments.DataSetSubscriptionDescriptor.CommunicationParameters.BrokerDataSetReaderTransportParameters.QueueName = "opcuademo/json"

            ' Instantiate the subscriber object and hook events.
            Dim subscriber = New EasyUASubscriber()
            AddHandler subscriber.DataSetMessage, AddressOf subscriber_DataSetMessage_MqttJsonTcp

            Console.WriteLine("Subscribing...")
            subscriber.SubscribeDataSet(subscribeDataSetArguments)

            Console.WriteLine("Processing dataset message events for 20 seconds...")
            Threading.Thread.Sleep(20 * 1000)

            Console.WriteLine("Unsubscribing...")
            subscriber.UnsubscribeAllDataSets()

            Console.WriteLine("Waiting for 1 second...")
            ' Unsubscribe operation is asynchronous, messages may still come for a short while.
            Threading.Thread.Sleep(1 * 1000)

            Console.WriteLine("Finished...")
        End Sub

        Private Shared Sub subscriber_DataSetMessage_MqttJsonTcp(ByVal sender As Object, ByVal e As EasyUADataSetMessageEventArgs)
            ' Display the dataset.
            If e.Succeeded Then
                ' An event with null DataSetData just indicates a successful connection.
                If e.DataSetData IsNot Nothing Then
                    Console.WriteLine()
                    Console.WriteLine($"Dataset data: {e.DataSetData}")
                    For Each pair As KeyValuePair(Of String, UADataSetFieldData) In e.DataSetData.FieldDataDictionary
                        Console.WriteLine(pair)
                    Next
                End If
            Else
                Console.WriteLine()
                Console.WriteLine($"*** Failure: {e.ErrorMessage}")
            End If
        End Sub

        ' Example output:
        '
        'Subscribing...
        'Processing dataset message events for 20 seconds...
        '
        'Dataset data: 2020-01-21T17:07:19.778,836,700,00; Good; Data; publisher=[String]31, class=eae79794-1af7-4f96-8401-4096cd1d8908, fields: 4
        '[BoolToggle, True {System.Boolean} @2020-01-21T16:07:19.778,836,700,00; Good]
        '[Int32, 482 {System.Int64} @2020-01-21T16:07:19.778,836,700,00; Good]
        '[Int32Fast, 2287 {System.Int64} @2020-01-21T16:07:19.778,836,700,00; Good]
        '[DateTime, 1/21/2020 5:07:19 PM {System.DateTime} @2020-01-21T16:07:19.778,836,700,00; Good]
        '
        'Dataset data: Good; Data; publisher=[String]32, fields: 4
        '[BoolToggle, True {System.Boolean}; Good]
        '[Int32, 482 {System.Int32}; Good]
        '[Int32Fast, 2287 {System.Int32}; Good]
        '[DateTime, 1/21/2020 5:07:19 PM {System.DateTime}; Good]
        '
        'Dataset data: Good; Data; publisher=[String]32, fields: 100
        '[Mass_0, 82 {System.Int64}; Good]
        '[Mass_1, 182 {System.Int64}; Good]
        '[Mass_2, 282 {System.Int64}; Good]
        '[Mass_3, 382 {System.Int64}; Good]
        '[Mass_4, 482 {System.Int64}; Good]
        '[Mass_5, 582 {System.Int64}; Good]
        '[Mass_6, 682 {System.Int64}; Good]
        '[Mass_7, 782 {System.Int64}; Good]
        '[Mass_8, 882 {System.Int64}; Good]
        '[Mass_9, 982 {System.Int64}; Good]
        '[Mass_10, 1082 {System.Int64}; Good]
        '...
    End Class
End Namespace

COM

REM The following package needs to be referenced in your project (or otherwise made available) for the MQTT transport to 
REM work.
REM - OpcLabs.MqttNet
REM Refer to the documentation for more information.
REM
REM Find all latest examples here : https://opclabs.doc-that.com/files/onlinedocs/OPCLabs-OpcStudio/Latest/examples.html .

' The subscriber object, with events
'Public WithEvents Subscriber4 As EasyUASubscriber

Private Sub EasyUASubscriber_SubscribeDataSet_MqttJsonTcp_Command_Click()
    OutputText = ""

    ' Define the PubSub connection we will work with. Uses implicit conversion from a string.
    ' Default port with MQTT is 1883.
    Dim subscribeDataSetArguments As New EasyUASubscribeDataSetArguments
    Dim pubSubConnectionDescriptor As UAPubSubConnectionDescriptor
    Set pubSubConnectionDescriptor = subscribeDataSetArguments.dataSetSubscriptionDescriptor.ConnectionDescriptor
    pubSubConnectionDescriptor.ResourceAddress.ResourceDescriptor.UrlString = "mqtt://opcua-pubsub.demo-this.com:1883"
    ' Specify the transport protocol mapping.
    ' The statement below isn't actually necessary, due to automatic message mapping recognition feature; see
    ' https://kb.opclabs.com/OPC_UA_PubSub_Automatic_Message_Mapping_Recognition for more details.
    pubSubConnectionDescriptor.TransportProfileUriString = "http://opcfoundation.org/UA-Profile/Transport/pubsub-mqtt-json" ' UAPubSubTransportProfileUriStrings.MqttJson

    ' Define the arguments for subscribing to the dataset, specifying the MQTT topic name.
    subscribeDataSetArguments.dataSetSubscriptionDescriptor.CommunicationParameters.BrokerDataSetReaderTransportParameters.QueueName = "opcuademo/json"
    
    ' Instantiate the subscriber object and hook events.
    Set Subscriber4 = New EasyUASubscriber
    
    OutputText = OutputText & "Subscribing..." & vbCrLf
    Call Subscriber4.SubscribeDataSet(subscribeDataSetArguments)

    OutputText = OutputText & "Processing dataset message for 20 seconds..." & vbCrLf
    Pause 20000

    OutputText = OutputText & "Unsubscribing..." & vbCrLf
    Subscriber4.UnsubscribeAllDataSets

    OutputText = OutputText & "Waiting for 1 second..." & vbCrLf
    ' Unsubscribe operation is asynchronous, messages may still come for a short while.
    Pause 1000

    Set Subscriber1 = Nothing

    OutputText = OutputText & "Finished." & vbCrLf
End Sub

Private Sub Subscriber4_DataSetMessage(ByVal sender As Variant, ByVal eventArgs As EasyUADataSetMessageEventArgs)
    ' Display the dataset
    If eventArgs.Succeeded Then
        ' An event with null DataSetData just indicates a successful connection.
        If Not eventArgs.DataSetData Is Nothing Then
            OutputText = OutputText & vbCrLf
            OutputText = OutputText & "Dataset data: " & eventArgs.DataSetData & vbCrLf
            Dim dictionaryEntry2 : For Each dictionaryEntry2 In eventArgs.DataSetData.FieldDataDictionary
                OutputText = OutputText & dictionaryEntry2 & vbCrLf
            Next
        End If
    Else
        OutputText = OutputText & vbCrLf
        OutputText = OutputText & eventArgs.ErrorMessageBrief & vbCrLf
    End If
End Sub
' Example output:
'
'Subscribing...
'Processing dataset message events for 20 seconds...
'
'Dataset data: 2020-01-21T17:07:19.778,836,700,00; Good; Data; publisher=[String]31, class=eae79794-1af7-4f96-8401-4096cd1d8908, fields: 4
'[BoolToggle, True {System.Boolean} @2020-01-21T16:07:19.778,836,700,00; Good]
'[Int32, 482 {System.Int64} @2020-01-21T16:07:19.778,836,700,00; Good]
'[Int32Fast, 2287 {System.Int64} @2020-01-21T16:07:19.778,836,700,00; Good]
'[DateTime, 1/21/2020 5:07:19 PM {System.DateTime} @2020-01-21T16:07:19.778,836,700,00; Good]
'
'Dataset data: Good; Data; publisher=[String]32, fields: 4
'[BoolToggle, True {System.Boolean}; Good]
'[Int32, 482 {System.Int32}; Good]
'[Int32Fast, 2287 {System.Int32}; Good]
'[DateTime, 1/21/2020 5:07:19 PM {System.DateTime}; Good]
'
'Dataset data: Good; Data; publisher=[String]32, fields: 100
'[Mass_0, 82 {System.Int64}; Good]
'[Mass_1, 182 {System.Int64}; Good]
'[Mass_2, 282 {System.Int64}; Good]
'[Mass_3, 382 {System.Int64}; Good]
'[Mass_4, 482 {System.Int64}; Good]
'[Mass_5, 582 {System.Int64}; Good]
'[Mass_6, 682 {System.Int64}; Good]
'[Mass_7, 782 {System.Int64}; Good]
'[Mass_8, 882 {System.Int64}; Good]
'[Mass_9, 982 {System.Int64}; Good]
'[Mass_10, 1082 {System.Int64}; Good]
'...

See Also